//ECE 454 Spring 2008 Assignment 1
//Macaranas, Joseph jjmacara
//Turner, Travis t2turner
#include "marshal.h"
#include <set>
#include <string>
#include <signal.h>
#include <queue>

void * serverthread(void * parm);       /* thread function prototype    */

bool checkTerminate( void * message);

#define MAXBUFLEN					100
pthread_mutex_t		thd_mutex;
int 				thd_count = 0;

set<string> registeredProcedures;

struct bread_data{   int	thread_id;   int  len;   char *message;};

void catchSIGINT(int signal) {

	cout << "Server de-registering functions from binder." << endl;
	char *binderHost;
	binderHost = getenv("BINDER_ADDRESS");
	if(binderHost == NULL) {
		cout << "BINDER_ADDRESS environment variable not set." << endl;
		exit(INVALID_HOST);
	}
	int port = atoi(getenv("BINDER_PORT"));

	struct  	hostent  *h;
	struct  	sockaddr_in binder_address;
	struct  	sockaddr_in server_address;     /* structure to hold this address */
	int     	sd, sd2;
	int     	n;
	int			tries;
	binder_msg 	*send_this = (binder_msg *) malloc(MAX_LENGTH);
	int			msgLength;
	int 		alen;
	void 		*message;

	memset((char *)&binder_address, 0, sizeof(binder_address));
	binder_address.sin_family = AF_INET;
	binder_address.sin_port = htons((u_short)port);


	h = gethostbyname(binderHost);
	if(((char *)h) == NULL)	{
		cerr << "Binder has invalid host name: " <<  binderHost << "." << endl << endl;
		exit(INVALID_HOST);
	}


	memcpy(&binder_address.sin_addr, h->h_addr, h->h_length);


	sd = socket(PF_INET, SOCK_DGRAM, 0);
	if (sd < 0){
	   	cerr << "Socket creation failed for sending de-register request failed. ";
		cerr << "Reason: " << strerror(errno) << endl << endl;
	   	exit(SOCKET_FAILURE);
	}


	send_this->requestType = SERVER_DEREGISTER;


	set<string>::iterator temp = registeredProcedures.begin();
	string *delimitedFunctionList = new string;


	while(temp != registeredProcedures.end()) {
		delimitedFunctionList->append(*temp);
		delimitedFunctionList->append("(");
		temp++;
	}

	send_this->name = (char *)malloc(delimitedFunctionList->length());

	strcpy(send_this->name, delimitedFunctionList->c_str());

	send_this->argTypes = (int *)malloc(sizeof(int));
   	send_this->argTypes[0] = 0;


   	message = malloc(MAX_LENGTH);


   	binderPackData(message, send_this, &msgLength);
   	free(send_this);

   	delete delimitedFunctionList;
   	message = realloc(message, msgLength);

   	n = sendto(sd, message, msgLength, 0,
			(struct sockaddr *) &binder_address, sizeof(struct sockaddr) );


	if (n < 0) {
		cout << "Server failed to send de-register request to binder." << strerror(errno) << endl;
		free(message);
		close(sd);
		exit(SEND_ERROR);
	}
	close(sd);
	free(message);


	//open a socket to listen for a response from the binder
	server_address.sin_port = htons((u_short)port)+1;
	server_address.sin_family = AF_INET;
	server_address.sin_addr.s_addr = INADDR_ANY;


	sd2 = socket(PF_INET, SOCK_DGRAM, 0);
	if (sd2 < 0){
		cerr << "Socket creation for receiving data from the binder failed. ";
		cerr << "Reason: " << strerror(errno) << endl << endl;
		exit(SOCKET_FAILURE);
	}


	void *reply = (void *)malloc(MAX_LENGTH);
	binder_msg *receive_this;
	alen = sizeof(struct sockaddr);

	/* Bind a local address to the socket */
	if (bind(sd2, (struct sockaddr *)&server_address, sizeof (server_address)) < 0) {
		cerr << "Bind for receiving data from the binder failed. ";
		cerr << "Reason: " << strerror(errno) << endl << endl;
		free(reply);
		exit(SOCKET_FAILURE);
	}

	if ((n = recvfrom(sd2, reply, MAX_LENGTH, 0,
							 (struct sockaddr *)&server_address, (socklen_t *)&alen)) < 0) {
		cerr << "Server failed to receive data from the binder. ";
		cerr << "Reason: " << strerror(errno) << endl << endl;
		close(sd2);
		free(reply);
		exit(RESULT_FAILURE);
	}

	reply = realloc(reply, n);
	receive_this = (binder_msg *)malloc(n);

	binderUnpackData(reply, receive_this);

	if (receive_this->requestType == REQUEST_SUCCESS) {
		cout << "Successfully de-registered functions ";
		cout << "with the binder." << endl << endl;
		//memory clean up
		close(sd2);
		free(reply);
		free(receive_this);
		//set handler for next sigint
		//(void) signal(SIGINT, catchSIGINT);
		exit(RESULT_SUCCESS);
	} else {
		cerr << "Error occured during de-registration request. ";
		cerr << "Terminating anyways..." << endl << endl;
		close(sd2);
		free(reply);
		free(receive_this);
		exit(RESULT_FAILURE);
	}

}

int rpcExecute(void) {

	//set handler for sigint
	signal(SIGINT, catchSIGINT);

	//initialize sockets, variables, threads
    struct   hostent   *ptrh;     /* pointer to a host table entry */
    struct   sockaddr_in recvaddr;     /* structure to hold server's address */
    struct   sockaddr_in cad;     /* structure to hold client's address */
    int      sd, sd2;             /* socket descriptors */
    int      port;                /* protocol port number */
    int      alen;                /* length of address */
	pthread_t  tid;               /* variable to hold thread ID */
    void 	*buf;
	int n;

	pthread_mutex_init(&thd_mutex, NULL);

    memset((char  *)&recvaddr,0,sizeof(recvaddr));
    recvaddr.sin_family = AF_INET;
    recvaddr.sin_addr.s_addr = INADDR_ANY;
	recvaddr.sin_port = htons((u_short)PORT);

     /* Create a socket */
    sd = socket (PF_INET, SOCK_DGRAM, 0);
    if (sd < 0) {
		cerr << "Socket creation for receiving RPC requests failed. ";
        cerr << "Reason: " << strerror(errno) << endl;
        return SOCKET_FAILURE;
    }

     /* Bind a local address to the socket */
    if (bind(sd, (struct sockaddr *)&recvaddr, sizeof (recvaddr)) < 0) {
        cerr << "Server bind for receiving RPC requests failed. ";
        cerr << "Reason: " << strerror(errno) << endl;
    	return SOCKET_FAILURE;
    }

	//wait for a request from a client
	alen = sizeof(struct sockaddr);

     /* Main server loop - accept and handle requests */
    cout << "Server ready to accept RPC requests via port: " << PORT << "." << endl;

    while (1) {

    	try {

			buf = malloc(MAX_LENGTH);

       		cout<< "Server blocking until a request is received." << endl;
       		if ((n = recvfrom(sd, buf, MAX_LENGTH, 0,
                         (struct sockaddr *)&cad, (socklen_t *)&alen)) < 0) {
				cerr << "Server failed to receive a message during a transmission. ";
				cerr << "Reason: " << strerror(errno) << endl << endl;
				free(buf);
				throw RECV_ERROR; //skip to next recv_from
	    	}

			cout << "Server received a message with size " << n << " bytes." << endl;

			//if we get the terminate message from binder
			//message conists of 320 consecutive 0's
			if(checkTerminate(buf)) {
				cout << "Server shutting down." << endl;
				close(sd);
				free(buf);
				return 0;
			}

			thread_data *param = (thread_data *)malloc(MAX_LENGTH * 2);

			//extract data from message and place in param struct
			rpcExtractData(buf, param);

			param->thread_id = thd_count;
			param->len = n;
			param->return_addr = cad;
			param->message = buf;

			//cannot support more than 2 concurrent RPC requests
			if (thd_count >= 2) {

				cerr << "Server is busy and rejects an RPC request." << endl;

				//send a failure message
				void *response = malloc(sizeof(int));
				memset(response, 0, sizeof(int));
				int sd1;

				param->return_addr.sin_family = AF_INET;
				param->return_addr.sin_port = htons((u_short)PORT);

				sd1 = socket(PF_INET, SOCK_DGRAM, 0);
				if (sd1 < 0) {
	    			cerr << "Socket creation failed for rejecing RPC request. ";
	    			cerr << "Reason: " << strerror(errno) << endl;
					free(buf);
					free(param);
					close(sd1);
					throw SOCKET_FAILURE;
				}

				int n1 = sendto(sd, response, sizeof(int), 0, (struct sockaddr *)&(param->return_addr), sizeof(struct sockaddr));

				if (n1 == -1) {
					cerr << "Error sending data to reject RPC request. ";
					cerr << "Reason: " << strerror(errno) << endl;
					close(sd1);
					free(buf);
					free(param);
					free(response);
					throw SOCKET_FAILURE;
				}

				close(sd1);
				free(buf);
				free(response);
				free(param);

			} else {
				cout << "Server creating a new thread to perform the remote procedure call." << endl;
				int val = pthread_create(&tid, NULL, serverthread, param);
				if ( val !=0 ){
					cerr << "Error creating thread for new RPC request. ";
					cerr << "Reason: " << val << endl;
					free(buf);
					free(param);
				}
				//cout << "Thread that was created for a request has died." << endl;
			}
		} catch (int e) {
			//do nothing, get next message
		}
	}
}

void * serverthread(void * parm) {

	cout << "New thread started." << endl;

	int     sd;
	struct thread_data *d = (struct thread_data *)parm;
	function f = d->f;
	int len = d->len;

	pthread_mutex_lock(&thd_mutex);
	thd_count++;
	pthread_mutex_unlock(&thd_mutex);

	cout << "Server now executing remotely-invoked procedure." << endl;

	d->f(d->argTypes, d->args);	//CALL FUNCTION

	cout << "Remote procedure successfully called." << endl;

	void *response = (void *)malloc(MAX_LENGTH);
	int responseLength = 0;

	if(rpcPackData(d->argTypes, d->args, response, &responseLength) == 0) {
		cout << "Server successfully packed data to send to client." << endl;
	} else {
		cout << "Error packing data for the response." << endl;
		pthread_exit(0);
	}

	d->return_addr.sin_family = AF_INET;
	d->return_addr.sin_port = htons((u_short)PORT + 1);

	sd = socket(PF_INET, SOCK_DGRAM, 0);
	if (sd < 0){
	    free(response);
	    free(d);
	    cerr << "Socket creation failed for the response." << endl;
	    pthread_mutex_lock(&thd_mutex);
		thd_count--;
		pthread_mutex_unlock(&thd_mutex);
	    pthread_exit(0);
	}

	int n = sendto(sd, response, responseLength, 0, (struct sockaddr *)&d->return_addr, sizeof(struct sockaddr));

	if (n == -1) {
		cerr << "Error sending response to the client." << endl;
	} else {
		cout << "Server successfully sent response to client with ";
		cout << n << " bytes." << endl;
	}

	close(sd);

	free(response);
	free(d);

	pthread_mutex_lock(&thd_mutex);
	thd_count--;
	pthread_mutex_unlock(&thd_mutex);
	pthread_exit(0);

}

//called by server to register a function with the binder
int rpcRegister(char *name, int *argTypes, function f) {

	char *binderHost;
	binderHost = getenv("BINDER_ADDRESS");

	if(binderHost == NULL) {		//ensure binder address is valid
		cout << "BINDER_ADDRESS environment variable not set." << endl;
		exit(INVALID_HOST);
	}

	cout << "Registering function " << name << " with the binder." << endl;

	int port = atoi(getenv("BINDER_PORT"));	//get the port

	struct  hostent  *h;
	struct  sockaddr_in binder_address;
	struct  sockaddr_in server_address;     /* structure to hold this address */
	int     sd, sd2;
	int     n;
	int		tries;
	binder_msg *send_this;
	binder_msg *receive_this;
	void 	*message;
	int		msgLength;
	int 	alen;

	memset((char *)&binder_address, 0, sizeof(binder_address));	//zero out the binder address
	binder_address.sin_family = AF_INET;
	binder_address.sin_port = htons((u_short)port);

	h = gethostbyname(binderHost);
	if(((char *)h) == NULL)	{		//bad host name
		cerr << "Binder has invalid host name: " <<  binderHost << "." << endl << endl;
		return INVALID_HOST;
	}

	memcpy(&binder_address.sin_addr, h->h_addr, h->h_length);	//copy address to binder_address

	//create a socket to send message to the binder
	sd = socket(PF_INET, SOCK_DGRAM, 0);
	if (sd < 0){
	   cerr << "Socket creation for sending a request to the binder failed. ";
	   cerr << "Reason: " << strerror(errno) << endl << endl;
	   return SOCKET_FAILURE;
	}

	//set up a binder msg struct to send the data
	send_this = (binder_msg *)malloc(MAX_LENGTH);

	send_this->requestType = SERVER_REGISTER;
	send_this->name = name;
	send_this->argTypes = argTypes;
	send_this->servers = NULL;

	function *tempArray = (function *)malloc(2 * sizeof(function));
	tempArray[0] = f;
	tempArray[1] = 0;

	send_this->functions = tempArray;

	message = malloc(MAX_LENGTH);

	//marshall the data into a message
	binderPackData(message, send_this, &msgLength);

	free(tempArray);

	message = realloc(message, msgLength);

	if (send_this->functions != NULL) {
		free(send_this->functions);
	}
	free(send_this);

	cout << "Server sending registration request to binder." << endl;

	//send the message
	n = sendto(sd, message, msgLength, 0,
		(struct sockaddr *) &binder_address, sizeof(struct sockaddr) );

	//something bad happened
	if (n < 0) {
		cout << "Server failed to transmit data to the binder." << endl << endl;
		close(sd);
		free(message);
		return SEND_ERROR;
	}
	close(sd);
	free(message);
	cout << "Server succesfully sent registration request to binder." << endl;
	cout << "Server waiting for a response from the binder." << endl;

	//open a socket to listen for a response from the binder
	server_address.sin_port = htons((u_short)port)+1;	//be sure to usse
	server_address.sin_family = AF_INET;
	server_address.sin_addr.s_addr = INADDR_ANY;

	sd2 = socket(PF_INET, SOCK_DGRAM, 0);
	if (sd2 < 0){
		cerr << "Socket creation for receiving data from the binder failed. ";
		cerr << "Reason: " << strerror(errno) << endl << endl;
		return SOCKET_FAILURE;
	}

	void *reply = (void *)malloc(MAX_LENGTH);
	alen = sizeof(struct sockaddr);

	/* Bind a local address to the socket */
	if (bind(sd2, (struct sockaddr *)&server_address, sizeof (server_address)) < 0) {
		cerr << "Bind for receiving data from the binder failed. ";
		cerr << "Reason: " << strerror(errno) << endl << endl;
		close(sd2);
		return SOCKET_FAILURE;
	}

	if ((n = recvfrom(sd2, reply, MAX_LENGTH, 0,
							 (struct sockaddr *)&server_address, (socklen_t *)&alen)) < 0) {
		cerr << "Server failed to receive data from the binder. ";
		cerr << "Reason: " << strerror(errno) << endl << endl;
		close(sd2);
		free(reply);
		free(receive_this);
		return RESULT_FAILURE;
	}

	receive_this = (binder_msg *)malloc(n);

	binderUnpackData(reply, receive_this);

	if(receive_this->requestType == REQUEST_SUCCESS) {
		cout << "Successfully registered function " << name;
		cout << " with the binder." << endl << endl;
		//add to set of registered procedures
		string function_name = name;
		registeredProcedures.insert(function_name);
		//memory clean up
		close(sd2);
		free(reply);
		free(receive_this);
		return RESULT_SUCCESS;
	} else if(receive_this->requestType == REQUEST_WARNING) {
		cout << "Function " << name << " with this set of argument ";
		cout << "types has already been registered by this server." << endl << endl;
		close(sd2);
		free(reply);
		free(receive_this);
		return RESULT_WARNING;
	} else {
		cerr << "Error occured during registration request." << endl << endl;
		close(sd2);
		free(reply);
		free(receive_this);
		return RESULT_FAILURE;
	}

}

//helper function to check a message for the terminate condition as passed
//by the binder.
//if server receives 10 consecutive INTs with their values set to 0, then the
//server must shut down
bool checkTerminate(void * message) {

	int i = 0;
	void * msgPtr = message;

	while(i < 10) {
		int *tempPtr = (int *)msgPtr;

		if(*tempPtr != 0) {
			return false;
		}
		tempPtr++;
		msgPtr = (void *)tempPtr;
		i++;
	}
	return true;
}
